package com.baishancloud.log.live.tengine

import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.lang

/**
 *
 * @author ziqiang.wang 
 * @date 2021/11/30 17:22
 */
class TengineWindowAgg(parameterTool: ParameterTool) extends ProcessWindowFunction[TengineLog, TengineResult, TengineLogKey, TimeWindow] {

  //状态未被修改最长保留时间
  var ttlHour: Long = _
  var logCount: ValueState[lang.Long] = _

  override def open(parameters: Configuration): Unit = {
    ttlHour = parameterTool.getLong(allowOldestHour, 6)
    val valueStateDescriptor: ValueStateDescriptor[lang.Long] = new ValueStateDescriptor[lang.Long]("日志条数", BasicTypeInfo.LONG_TYPE_INFO)
    val ttlConfig: StateTtlConfig = StateTtlConfig
      .newBuilder(Time.hours(ttlHour))
      .updateTtlOnCreateAndWrite()
      .returnExpiredIfNotCleanedUp()
      .useProcessingTime()
      .updateTtlOnCreateAndWrite()
      .build()
    valueStateDescriptor.enableTimeToLive(ttlConfig)
    logCount = getRuntimeContext.getState(valueStateDescriptor)
  }

  override def process(key: TengineLogKey, context: Context, elements: Iterable[TengineLog], out: Collector[TengineResult]): Unit = {
    if (logCount.value() == null) {
      logCount.update(0)
    }
    logCount.update(logCount.value() + elements.size)
    out.collect(TengineResult(key.time_at_1min, key.domain, key.country, key.province, key.isp, key.log_type, logCount.value()))
  }
}
